package com.huan.filter.udf;

import com.huan.filter.vo.Person;
import org.apache.flink.api.common.functions.FilterFunction;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * Flink暴露了所有UDF函数的接口，具体实现方式为接口或者抽象类，例如MapFunction、FilterFunction、ReduceFunction等。所以用户可以自定义一个函数类，实现对应的接口。
 *
 * @author huan.fu
 * @date 2023/9/18 - 22:43
 */
public class UdfFilterApplication {

    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment environment = StreamExecutionEnvironment.getExecutionEnvironment();

        environment.fromElements(
                        new Person(1, "张三", 20),
                        new Person(2, "李四", 25),
                        new Person(3, "王五", 30)
                )
                // filter 过滤
                .filter(new AgeFilter())
                .print();

        environment.execute("map operation");
    }

    public static class AgeFilter implements FilterFunction<Person> {

        @Override
        public boolean filter(Person person) throws Exception {
            return person.getAge() > 20;
        }
    }
}